package fr.neyb.wesb.core.job.impl;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import fr.neyb.framework.exceptions.TechnicalException;
import fr.neyb.framework.observer.Observable;
import fr.neyb.framework.observer.impl.AbstractObservable;
import fr.neyb.wesb.core.data.descriptors.record.RecordDescriptor;
import fr.neyb.wesb.core.data.io.RecordIO;
import fr.neyb.wesb.core.data.io.RecordReader;
import fr.neyb.wesb.core.data.io.RecordWriter;
import fr.neyb.wesb.core.job.Job;
import fr.neyb.wesb.core.job.RunningState;
import fr.neyb.wesb.core.job.runner.JobsRunner;

public abstract class AbstractJob extends AbstractObservable implements Job {

	RunningState currentState = RunningState.NOT_INITIALIZED;

	protected final List<RecordDescriptor> inputsDescriptors;
	protected final List<RecordDescriptor> outputsDescriptors;

	final List<RecordReader> readers;
	final List<RecordWriter> writers;

	private final Set<Job> dependencies = new HashSet<>();

	public AbstractJob(List<RecordDescriptor> inputsDescriptors,
			List<RecordDescriptor> outputsDescriptors) {
		this.inputsDescriptors = Collections.unmodifiableList(new ArrayList<>(
				inputsDescriptors));
		this.outputsDescriptors = Collections.unmodifiableList(new ArrayList<>(
				outputsDescriptors));
		RecordReader[] readers = new RecordReader[inputsDescriptors.size()];
		RecordWriter[] writers = new RecordWriter[outputsDescriptors.size()];
		Arrays.fill(readers, null);
		Arrays.fill(writers, null);
		this.readers = Arrays.asList(readers);
		this.writers = Arrays.asList(writers);
		if (inputsDescriptors.size() == 0)
			currentState = RunningState.READY;
	}

	@Override
	public synchronized boolean setReader(int i, RecordReader reader) {
		return setIO(i, reader, readers, inputsDescriptors);
	}

	@Override
	public synchronized boolean setWriter(int i, RecordWriter writer) {
		return setIO(i, writer, writers, outputsDescriptors);
	}

	@Override
	public synchronized boolean addDependency(Job dependency) {
		boolean result = false;
		dependencies.add(dependency);
		if (result && dependency.getState() != RunningState.ENDED) {
			//FIXME What if dependency.getState() == ENDED now ??? (multithreading)
			currentState = RunningState.WAITING;
		}
		return dependencies.add(dependency);
	}

	@Override
	public RunningState getState() {
		return currentState;
	}

	@Override
	public synchronized void start(JobsRunner runner) {
		updateState();
		if (currentState == RunningState.READY
				|| currentState == RunningState.WAITING) {
			currentState = RunningState.READY;
			for (Job dep : dependencies) {
				if (dep.getState() != RunningState.ENDED) {
					currentState = RunningState.WAITING;
					dep.addObserver(this);
				} else {
					dep.deleteObserver(this);
				}
			}
		}
		if (currentState == RunningState.READY) {
			runner.registerForRun(this);
		}
	}

	@Override
	public void update(Observable o, Object arg) {
		if (o instanceof Job) {
			if (arg instanceof JobsRunner) {
				start((JobsRunner) arg);
			} else {
				throw new TechnicalException("");// TODO message
			}

		}
	}

	private <T extends RecordIO> boolean setIO(int i, T recIO, List<T> list,
			List<RecordDescriptor> descr) {
		if (recIO == null || recIO.getDescriptor().equals(descr.get(i))) {
			T oldRecIO = list.set(i, recIO);
			boolean result = oldRecIO == null ? false : oldRecIO.equals(recIO);
			updateState();
			return result;
		}
		return false;
	}

	private synchronized RunningState updateState() {
		if (canMove()) {
			for (RecordReader rr : readers) {
				if (rr == null) {
					currentState = RunningState.NOT_INITIALIZED;
					return currentState;
				}
			}
			for (RecordWriter rw : writers) {
				if (rw == null) {
					currentState = RunningState.NOT_INITIALIZED;
					return currentState;
				}
			}

			currentState = RunningState.READY;
			return currentState;
		}
		return currentState;
	}

	private boolean canMove() {
		return currentState.compareTo(RunningState.READY) <= 0;
	}

	@Override
	public void setState(RunningState newState) {
		if (newState.compareTo(currentState) > 0) {
			currentState = newState;
			setChanged();
		}
		if (currentState == RunningState.ENDED)
			notifyObservers();
	}

	protected void End(JobsRunner runner) {
		if (currentState != RunningState.ENDED)
			setChanged();
		currentState = RunningState.ENDED;
		notifyObservers(runner);
	}
}
